package com.sui.bigdata.flink.sql.function.aggregate;

import org.apache.flink.table.functions.AggregateFunction;
import com.sui.bigdata.flink.sql.function.util.PhoneHHIAccum;
import com.sui.bigdata.flink.sql.function.scalar.NotBlank;

import java.math.BigDecimal;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.sui.bigdata.flink.sql.function.util.Utils.isMobileNO;

/**
 * @author caokaizhi
 * @date 2019-09-02 12:35
 * @description
 * @reviewer
 */
public class PhoneHHI extends AggregateFunction<Double, PhoneHHIAccum> {
    private static Logger logger = LoggerFactory.getLogger(PhoneHHI.class);

    @Override
    public PhoneHHIAccum createAccumulator() {
        return new PhoneHHIAccum();
    }

    @Override
    public Double getValue(PhoneHHIAccum phoneHHIAccum) {
        Double result = 0.0000;
        if (phoneHHIAccum != null && phoneHHIAccum.getCount() != 0 && !phoneHHIAccum.getGroups().isEmpty()) {
            Set<Map.Entry<String, Integer>> entries = phoneHHIAccum.getGroups().entrySet();
            for (Map.Entry<String, Integer> entry : entries) {
                result += Math.pow((double) entry.getValue() / phoneHHIAccum.getCount(), 2);
            }
        }
        BigDecimal bigDecimal = new BigDecimal(result);
        Double a = bigDecimal.setScale(4, BigDecimal.ROUND_HALF_UP).doubleValue();
        return a;
    }

    public void accumulate(PhoneHHIAccum phoneHHIAccum, String phone, Integer start, Integer end) {
        if (NotBlank.eval(phone) && !phoneHHIAccum.exitsPhone(phone) && isMobileNO(phone)) {
            phoneHHIAccum.addphone(phone);
            phoneHHIAccum.setCount(phoneHHIAccum.getCount() + 1);
            if ("0".equals(phone.substring(0, 1)) && phone.length() == 11) {
                phone = phone.substring(3, 11);
            } else if ("0".equals(phone.substring(0, 1)) && phone.length() == 12) {
                phone = phone.substring(4, 12);
            }
            if (phone.length() >= end) {
                subString(phoneHHIAccum, phone, start, end);
            }
        }
    }

    public void accumulate(PhoneHHIAccum phoneHHIAccum, String phone, Integer start) {
        if (NotBlank.eval(phone) && !phoneHHIAccum.exitsPhone(phone) && isMobileNO(phone)) {
            phoneHHIAccum.addphone(phone);
            phoneHHIAccum.setCount(phoneHHIAccum.getCount() + 1);
            if ("0".equals(phone.substring(0, 1)) && phone.length() == 11) {
                phone = phone.substring(3, 11);
            } else if ("0".equals(phone.substring(0, 1)) && phone.length() == 12) {
                phone = phone.substring(4, 12);
            }
            if (start == 4 && phone.length() >= 6) {
//                Integer s = Math.toIntExact(Math.round((double) ((phone.length() - 4) / 2))) + phone.length() % 2;
                Integer s = (int) (Math.ceil((float) phone.length() / 2) - 2);
                subString(phoneHHIAccum, phone, s, s + 4);
            } else if (start == -4 && phone.length() >= 6) {
//                phoneHHIAccum.setCount(phoneHHIAccum.getCount() + 1);
                String substring = phone.substring(phone.length() - 4);
                if (phoneHHIAccum.getGroups().get(substring) == null) {
                    phoneHHIAccum.getGroups().put(substring, 1);
                } else {
                    phoneHHIAccum.getGroups().put(substring, phoneHHIAccum.getGroups().get(substring) + 1);
                }
            }
        }
    }


    public void subString(PhoneHHIAccum phoneHHIAccum, String phone, Integer start, Integer end) {
//        phoneHHIAccum.setCount(phoneHHIAccum.getCount() + 1);
        String substring = phone.substring(start, end);
        if (phoneHHIAccum.getGroups().get(substring) == null) {
            phoneHHIAccum.getGroups().put(substring, 1);
        } else {
            phoneHHIAccum.getGroups().put(substring, phoneHHIAccum.getGroups().get(substring) + 1);
        }
    }

    public void merge(PhoneHHIAccum phoneHHIAccum, Iterable<PhoneHHIAccum> it) {
        Iterator<PhoneHHIAccum> iter = it.iterator();
        while (iter.hasNext()) {
            PhoneHHIAccum a = iter.next();
            if (phoneHHIAccum.getGroups().isEmpty()) {
                phoneHHIAccum.getGroups().putAll(a.getGroups());

            } else {
                for (Map.Entry<String, Integer> entry : a.getGroups().entrySet()) {
                    if (phoneHHIAccum.getGroups().get(entry.getKey()) == null) {
                        phoneHHIAccum.getGroups().put(entry.getKey(), entry.getValue());
                    } else {
                        phoneHHIAccum.getGroups().put(entry.getKey(), entry.getValue() + phoneHHIAccum.getGroups().get(entry.getKey()));
                    }
                }
            }

            phoneHHIAccum.setCount(phoneHHIAccum.getCount() + a.getCount());
            phoneHHIAccum.getRepetition().addAll(a.getRepetition());
        }
    }

}
